/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.mapred;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType;
import org.apache.hadoop.mapreduce.server.jobtracker.JTConfig;

/**
 * A {@link JobInProgressListener} which initializes the tasks for a job as soon
 * as the job is added (using the {@link #jobAdded(JobInProgress)} method).
 */
class EagerTaskInitializationListener extends JobInProgressListener {

	class InitJob implements Runnable {

		private JobInProgress job;

		public InitJob(JobInProgress job) {
			this.job = job;
		}

		public void run() {
			ttm.initJob(job);
		}
	}
	// ///////////////////////////////////////////////////////////////
	// Used to init new jobs that have just been created
	// ///////////////////////////////////////////////////////////////
	class JobInitManager implements Runnable {

		public void run() {
			JobInProgress job = null;
			while (true) {
				try {
					synchronized (jobInitQueue) {
						while (jobInitQueue.isEmpty()) {
							jobInitQueue.wait();
						}
						job = jobInitQueue.remove(0);
					}
					threadPool.execute(new InitJob(job));
				} catch (InterruptedException t) {
					LOG.info("JobInitManagerThread interrupted.");
					break;
				}
			}
			LOG.info("Shutting down thread pool");
			threadPool.shutdownNow();
		}
	}

	private static final int DEFAULT_NUM_THREADS = 4;

	private static final Log LOG = LogFactory
			.getLog(EagerTaskInitializationListener.class.getName());

	private JobInitManager jobInitManager = new JobInitManager();
	private Thread jobInitManagerThread;
	private List<JobInProgress> jobInitQueue = new ArrayList<JobInProgress>();
	private ExecutorService threadPool;
	private int numThreads;
	TaskTrackerManager ttm;

	public EagerTaskInitializationListener(Configuration conf) {
		numThreads = conf.getInt(JTConfig.JT_JOBINIT_THREADS,
				DEFAULT_NUM_THREADS);
		threadPool = Executors.newFixedThreadPool(numThreads);
	}

	/**
	 * We add the JIP to the jobInitQueue, which is processed asynchronously to
	 * handle split-computation and build up the right TaskTracker/Block
	 * mapping.
	 */
	@Override
	public void jobAdded(JobInProgress job) {
		synchronized (jobInitQueue) {
			jobInitQueue.add(job);
			resortInitQueue();
			jobInitQueue.notifyAll();
		}

	}

	@Override
	public void jobRemoved(JobInProgress job) {
		synchronized (jobInitQueue) {
			jobInitQueue.remove(job);
		}
	}

	// called when the job's status is changed
	private void jobStateChanged(JobStatusChangeEvent event) {
		// Resort the job queue if the job-start-time or job-priority changes
		if (event.getEventType() == EventType.START_TIME_CHANGED
				|| event.getEventType() == EventType.PRIORITY_CHANGED) {
			synchronized (jobInitQueue) {
				resortInitQueue();
			}
		}
	}

	@Override
	public void jobUpdated(JobChangeEvent event) {
		if (event instanceof JobStatusChangeEvent) {
			jobStateChanged((JobStatusChangeEvent) event);
		}
	}

	/**
	 * Sort jobs by priority and then by start time.
	 */
	private synchronized void resortInitQueue() {
		Comparator<JobInProgress> comp = new Comparator<JobInProgress>() {
			public int compare(JobInProgress o1, JobInProgress o2) {
				int res = o1.getPriority().compareTo(o2.getPriority());
				if (res == 0) {
					if (o1.getStartTime() < o2.getStartTime())
						res = -1;
					else
						res = (o1.getStartTime() == o2.getStartTime() ? 0 : 1);
				}

				return res;
			}
		};

		synchronized (jobInitQueue) {
			Collections.sort(jobInitQueue, comp);
		}
	}

	public void setTaskTrackerManager(TaskTrackerManager ttm) {
		this.ttm = ttm;
	}

	public void start() throws IOException {
		this.jobInitManagerThread = new Thread(jobInitManager, "jobInitManager");
		jobInitManagerThread.setDaemon(true);
		this.jobInitManagerThread.start();
	}

	public void terminate() throws IOException {
		if (jobInitManagerThread != null && jobInitManagerThread.isAlive()) {
			LOG.info("Stopping Job Init Manager thread");
			jobInitManagerThread.interrupt();
			try {
				jobInitManagerThread.join();
			} catch (InterruptedException ex) {
				ex.printStackTrace();
			}
		}
	}

}
